package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo10Union {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("Demo10Union")


    val sc = new SparkContext(conf)

    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8))
    val rdd2: RDD[Int] = sc.makeRDD(List(4, 5, 6, 7, 8, 9, 10))

    /**
      * union: 合并两个rdd, 两个rdd的数据类型要一致
      * union只是代码层面的合并，在物理层面没有合并
      * union 不会产生shuffle
      *
      */
    val unionRDD: RDD[Int] = rdd1.union(rdd2)

    unionRDD.foreach(println)

    /**
      * distinct： 去重
      * distinct : 会产生shuffle
      * distinct； 会现在map端局部去重，再到reduce全局去重
      */

    val distinctRDD: RDD[Int] = unionRDD.distinct()

    distinctRDD.foreach(println)

    /**
      * 所有会产生shuffle的算子都可以指定分区数
      *
      */

    /**
      * intersection： 取两个rdd的交集
      *
      */

    val intersectionRDD: RDD[Int] = rdd1.intersection(rdd2)

    intersectionRDD.foreach(println)


  }

}
